package com.atguigu.tingshu.user.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.common.util.MongoUtil;
import com.atguigu.tingshu.model.user.UserListenProcess;
import com.atguigu.tingshu.user.service.UserListenProcessService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.user.UserListenProcessVo;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.fasterxml.jackson.databind.util.BeanUtil;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.temporal.ChronoUnit;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class UserListenProcessServiceImpl implements UserListenProcessService {

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private KafkaService kafkaService;

    @Override
    public BigDecimal getTrackBreakSecond(Long trackId) {
        Long userId = AuthContextHolder.getUserId();
        if (userId == null) {
            return new BigDecimal(0);
        }
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        UserListenProcess userListenProcess = this.mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId).and("trackId").is(trackId)), UserListenProcess.class, collectionName);
        if (userListenProcess == null) {
            return new BigDecimal(0);
        }
        return userListenProcess.getBreakSecond();
    }

    @Override
    public void updateListenProcess(UserListenProcessVo userListenProcessVo) {
        //1.判断用户是否存在
        Long userId = AuthContextHolder.getUserId();
        if (userId == null) {
            return;
        }
        //2.判断是更新还是新增 查询Mongodb
        String collectionName = MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId);
        UserListenProcess userListenProcess = this.mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId).and("trackId").is(userListenProcessVo.getTrackId()))
                , UserListenProcess.class, collectionName);
        if (userListenProcess == null) {
            //新增
            userListenProcess = new UserListenProcess();
            BeanUtils.copyProperties(userListenProcessVo, userListenProcess);
            userListenProcess.setUserId(userId);
            userListenProcess.setId(IdWorker.getIdStr());
            userListenProcess.setIsShow(1);
            userListenProcess.setCreateTime(new Date());
            userListenProcess.setUpdateTime(userListenProcess.getCreateTime());
        } else {
            //修改
            userListenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
            userListenProcess.setUpdateTime(new Date());
        }
        //存在就修改，不存在就新增
        this.mongoTemplate.save(userListenProcess, collectionName);
        //3.保证数据库声音以及专辑表修改 保证四个一 使用位图实现
        //获取到bit为true 直接返回
        Boolean bit = this.redisTemplate.opsForValue().getBit(RedisConstant.TRACK_STAT_PREFIX + userId, userListenProcessVo.getTrackId());
        if(bit){
            return ;
        }
        //bit为false 为添加记录
        this.redisTemplate.opsForValue().setBit(RedisConstant.TRACK_STAT_PREFIX + userId,userListenProcessVo.getTrackId(),true);
        //设置过期时间 每天的零点
        LocalDateTime endTime = LocalDateTime.of(LocalDate.now().plusDays(1), LocalTime.MIN);
        this.redisTemplate.expire(RedisConstant.TRACK_STAT_PREFIX + userId, ChronoUnit.SECONDS.between(LocalDateTime.now(),endTime), TimeUnit.SECONDS);
        //通过kafka发送消息
        try {
            TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
            trackStatMqVo.setStatType(SystemConstant.TRACK_STAT_PLAY);
            trackStatMqVo.setAlbumId(userListenProcessVo.getAlbumId());
            trackStatMqVo.setCount(1);
            trackStatMqVo.setTarckId(userListenProcessVo.getTrackId());
            trackStatMqVo.setBusinessNo(IdWorker.getIdStr());
            kafkaService.sendMsg(KafkaConstant.TRACK_ALBUM_STAT_INCR, JSON.toJSONString(trackStatMqVo));
        } catch (Exception e) {
            this.redisTemplate.opsForValue().setBit(RedisConstant.TRACK_STAT_PREFIX + userId,userListenProcessVo.getTrackId(),false);
            throw new RuntimeException(e);
        }
    }

    @Override
    public UserListenProcess getLatelyTrack() {
        Long userId = AuthContextHolder.getUserId();
        if(userId == null){
            return null;
        }
        UserListenProcess userListenProcess = this.mongoTemplate.findOne(Query.query(Criteria.where("userId").is(userId)).with(Sort.by(Sort.Order.desc("updateTime"))), UserListenProcess.class, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
        if(userListenProcess == null){
            return null;
        }
        return userListenProcess;
    }
}
